package com.atguigu.Flink.state;

import com.atguigu.Flink.POJO.Event;
import com.atguigu.Flink.function.ClickSource;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

public class Flink04_OperatorBroadCastState {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Event> dataDs = env.addSource(new ClickSource());
        MapStateDescriptor<String, String> mapStateDescriptor = new MapStateDescriptor<>("UNION", Types.STRING, Types.STRING);
        BroadcastStream<String> confDs = env.socketTextStream("hadoop102", 8888)
                .broadcast(
                        mapStateDescriptor
                );
        //数据流和广播流进行connect
        BroadcastConnectedStream<Event, String> bccs = dataDs.connect(confDs);
        bccs.process(
                new BroadcastProcessFunction<Event, String, String>() {
                    @Override
                    public void processElement(Event event, BroadcastProcessFunction<Event, String, String>.ReadOnlyContext readOnlyContext, Collector<String> collector) throws Exception {
                        ReadOnlyBroadcastState<String, String> broadcastState = readOnlyContext.getBroadcastState(mapStateDescriptor);
                        String conf = broadcastState.get("conf");
                        if("1".equals(conf)){
                            collector.collect("111111111111111");
                        }else if("2".equals(conf)){
                            collector.collect("2222222222222222");
                        }else{
                            collector.collect("默认啊！！！！！！！！！！！！！！！"+event);
                        }

                    }

                    @Override
                    public void processBroadcastElement(String s, BroadcastProcessFunction<Event, String, String>.Context context, Collector<String> collector) throws Exception {
                        BroadcastState<String, String> broadcastState = context.getBroadcastState(mapStateDescriptor);
                        broadcastState.put("conf",s);

                    }
                }
        ).print();


        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
